package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.func.DimAsyncFunction;
import com.atguigu.gmall.realtime.app.func.DimJoinFunction;
import com.atguigu.gmall.realtime.bean.OrderWide;
import com.atguigu.gmall.realtime.bean.PaymentWide;
import com.atguigu.gmall.realtime.bean.ProductStats;
import com.atguigu.gmall.realtime.common.GmallConstant;
import com.atguigu.gmall.realtime.utils.ClickHouseUtil;
import com.atguigu.gmall.realtime.utils.DateTimeUtil;
import com.atguigu.gmall.realtime.utils.MyKafka;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.text.ParseException;
import java.time.Duration;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

//商品主题统计，然后写入clickhouse
public class ProductStatsApp {
    public static void main(String[] args) throws Exception {
        //TODO 1 基本流环境的准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.1 设置并行度
        env.setParallelism(4);
        //设置checkpoint的的配置
        //设置没5s保存一次checkpoint,精准一次消费
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        //checkpoint必须在一分钟内完成，否则舍弃,检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(6000);
        //checkpoint保存的位置，及其操作用户
        env.setStateBackend(new FsStateBackend("hdfs://hadoop104:8020/gmall/flink/checkpoint2"));
        System.setProperty("HADOOP_USER_NAME","atguigu");
        //重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));

        //TODO 2 从kafka中读取数据
        //2.1根据ProductStatus实体类的属性，我们可以从以下实体类中读取数据
        String groupId="ProductStatsApp";
        //页面主题，里面可以获得page_id，当时details时候，就是进行了一次商品的点击，sku,并且也可以获得曝光数据
        String pageTopic="dwd_topic_page";
        //分流后写入到dwd层的kafak的收藏topic
        String  favorTopic="dwd_favor_info";
        //购物车
        String cartTopic="dwd_cart_info";
        //订单信息,从订单宽表中
        String orderWideTopic="dwm_order_wide";
        //评论
        String commentTopic="dwd_comment_info";
        //退款
        String refundTopic="dwd_order_refund_info";
        //支付
        String paymentTopic="dwm_payment_wide";
        FlinkKafkaConsumer<String> pageSource = MyKafka.getFlinkKafkaConsumer(pageTopic, groupId);
        FlinkKafkaConsumer<String> favorSource = MyKafka.getFlinkKafkaConsumer(favorTopic,groupId);
        FlinkKafkaConsumer<String> cartSource = MyKafka.getFlinkKafkaConsumer(cartTopic, groupId);
        FlinkKafkaConsumer<String> orderWideSource = MyKafka.getFlinkKafkaConsumer(orderWideTopic, groupId);
        FlinkKafkaConsumer<String> commentSource = MyKafka.getFlinkKafkaConsumer(commentTopic, groupId);
        FlinkKafkaConsumer<String> refundSource = MyKafka.getFlinkKafkaConsumer(refundTopic, groupId);
        FlinkKafkaConsumer<String> paymentSource = MyKafka.getFlinkKafkaConsumer(paymentTopic, groupId);

        //页面流
        DataStreamSource<String> pageDStream = env.addSource(pageSource);
        //收藏流
        DataStreamSource<String> favorDStream = env.addSource(favorSource);
        //购物车流
        DataStreamSource<String> cartDStream = env.addSource(cartSource);
        //订单流
        DataStreamSource<String> orderWideDStream = env.addSource(orderWideSource);
        //评论流
        DataStreamSource<String> commentDStream = env.addSource(commentSource);
        //支付流
        DataStreamSource<String> paymentDStream = env.addSource(paymentSource);
        //退款流
        DataStreamSource<String> refundDStream = env.addSource(refundSource);


        //TODO 3 将读取的数据进行结构转换
        //3.1 页面流,不使用map的原因是因为有曝光数据为集合,就是将数据分开处理后，发往下游
        SingleOutputStreamOperator<ProductStats> clickAndDisplayCtDS = pageDStream.process(
                new ProcessFunction<String, ProductStats>() {
                    @Override
                    public void processElement(String jsonStr, Context ctx, Collector<ProductStats> out) throws Exception {
                        //先将数据转换为json对象,这样可以拿到流中的具体数据
                        JSONObject jsonObject = JSON.parseObject(jsonStr);
                        //获取他的pageId "page_id": "good_detail", 可以查看是否为点击
                        String pageId = jsonObject.getJSONObject("page").getString("page_id");
                        //获取日志时间
                        Long ts = jsonObject.getLong("ts");
                        //进行判断
                        if (pageId.equals("good_detail")) {
                            //创建实体类对象，将点击写进去,并且也要写入item，确定是那个商品的点击
                            Long SkuId = jsonObject.getJSONObject("page").getLong("item");
                            ProductStats productStats = ProductStats.builder()
                                    .sku_id(SkuId)
                                    .click_ct(1L)
                                    .ts(ts)
                                    .build();
                            //将符合的数据发往下游
                            out.collect(productStats);
                        }

                        //曝光数据"item_type":"activity_id"  "item_type": "sku_id",当等于sku_Id时候就是曝光
                        //获取曝光数据
                        JSONArray displays = jsonObject.getJSONArray("displays");
                        if(displays!=null&& displays.size()>0){
                            for(int i=0;i<displays.size();i++){
                                JSONObject displaysJsonObj = displays.getJSONObject(i);
                                String isSkuId = displaysJsonObj.getString("item_type");
                                if (isSkuId.equals("sku_id")) {
                                    //获取 item,skuid,ts
                                    Long SkuId = displaysJsonObj.getLong("item");
                                    ProductStats productStats = ProductStats.builder()
                                            .sku_id(SkuId)
                                            .display_ct(1L)
                                            .ts(ts)
                                            .build();
                                    out.collect(productStats);
                                }
                            }

                        }



                    }
                }

        );

        //clickAndDisplayCtDS.print(">>");
        //3.2 收藏流
        SingleOutputStreamOperator<ProductStats> favorCtStatusDS = favorDStream.map(
                new MapFunction<String, ProductStats>() {
                    @Override
                    public ProductStats map(String jsonStr) throws Exception {
                        //结构转换
                        JSONObject jsonObject = JSON.parseObject(jsonStr);
                        Long skuId = jsonObject.getLong("sku_id");
                        String createTime = jsonObject.getString("create_time");
                        ProductStats productStas = ProductStats.builder()
                                .sku_id(skuId)
                                .favor_ct(1L)
                                .ts(DateTimeUtil.toTs(createTime))
                                .build();
                        return productStas;
                    }
                }
        );
        //favorCtStatusDS.print(">>");
        //3.3 购物车流
        SingleOutputStreamOperator<ProductStats> cartInfODS = cartDStream.map(
                new MapFunction<String, ProductStats>() {
                    @Override
                    public ProductStats map(String jsonStr) throws Exception {
                        JSONObject jsonObj = JSON.parseObject(jsonStr);
                        Long skuId = jsonObj.getLong("sku_id");
                        String createTime = jsonObj.getString("create_time");

                        return ProductStats.builder()
                                .sku_id(skuId)
                                .cart_ct(1L)
                                .ts(DateTimeUtil.toTs(createTime))
                                .build();
                    }
                }
        );
        // 3.4订单流
        SingleOutputStreamOperator<ProductStats> orderInfoCtDS = orderWideDStream.map(
                new MapFunction<String, ProductStats>() {
                    @Override
                    public ProductStats map(String jsonStr) throws Exception {
                        OrderWide orderWide = JSON.parseObject(jsonStr, OrderWide.class);
                        long ts = DateTimeUtil.toTs(orderWide.getCreate_time());
                        return ProductStats.builder()
                                .sku_id(orderWide.getSku_id())
                                .order_sku_num(orderWide.getSku_num())
                                //商品实付分摊之后的金额
                                .order_amount(orderWide.getSplit_total_amount())
                                //注意：因为数据是订单明细和订单的宽表，有可能多条明细对应同一个订单
                                // 所以在统计商品下单数的时候， 需要放到Set集合中去重
                                .orderIdSet(new HashSet(Collections.singleton(orderWide.getOrder_id())))
                                .ts(ts)
                                .build();
                    }
                }
        );
        //3.5 评论流
        SingleOutputStreamOperator<ProductStats> commentInfoDS = commentDStream.map(
                new MapFunction<String, ProductStats>() {

                    @Override
                    public ProductStats map(String jsonStr) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(jsonStr);
                        Long goodCt = GmallConstant.APPRAISE_GOOD.equals(jsonObject.getString("appraise")) ? 1L : 0L;
                        String create_time = jsonObject.getString("create_time");
                        long ts = DateTimeUtil.toTs(create_time);

                        ProductStats productStats = ProductStats.builder()
                                .ts(ts)
                                .comment_ct(1L)
                                .good_comment_ct(goodCt)
                                .sku_id(jsonObject.getLong("sku_id"))
                                .build();
                        return productStats;

                    }
                }
        );
        //paymentDStream.print(">>");
        //3.6支付流
        SingleOutputStreamOperator<ProductStats> paymentInfoDS = paymentDStream.map(
                json -> {
                    PaymentWide paymentWide = JSON.parseObject(json, PaymentWide.class);
                    String payment_create_time = paymentWide.getPayment_create_time();
                    if(payment_create_time !=null){
                        System.out.println("@time>>@"+payment_create_time);
                    }
                    System.out.println("@time@"+payment_create_time);
                    Long ts = DateTimeUtil.toTs(payment_create_time);
                    return ProductStats.builder()
                            .sku_id(paymentWide.getSku_id())
                            .payment_amount(paymentWide.getSplit_total_amount())
                            .paidOrderIdSet(new HashSet(Collections.singleton(paymentWide.getOrder_id())))
                            .ts(ts)
                            .build();
                }
        );
        //3.7退款流
        SingleOutputStreamOperator<ProductStats> refundInfoDS = refundDStream.map(
                json -> {
                    JSONObject refundJsonObj = JSON.parseObject(json);
                    Long ts = DateTimeUtil.toTs(refundJsonObj.getString("create_time"));
                    ProductStats productStats = ProductStats.builder()
                            .sku_id(refundJsonObj.getLong("sku_id"))
                            .refund_amount(refundJsonObj.getBigDecimal("refund_amount"))
                            .refundOrderIdSet(
                                    new HashSet(Collections.singleton(refundJsonObj.getLong("order_id"))))
                            .ts(ts).build();
                    return productStats;

                }
        );
        //TODO 4 将转换后的数据进行union
        DataStream<ProductStats> unionDS = clickAndDisplayCtDS.union(
                favorCtStatusDS,
                cartInfODS,
                orderInfoCtDS,
                commentInfoDS,
                paymentInfoDS,
                refundInfoDS
        );



        //TODO 5 设置水位线
        SingleOutputStreamOperator<ProductStats> productStatsWithWaterMarkDS = unionDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<ProductStats>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<ProductStats>() {
                            @Override
                            public long extractTimestamp(ProductStats productStats, long recordTimestamp) {
                                return productStats.getTs();
                            }
                        })
        );
        //TODO 6 分组
        KeyedStream<ProductStats, Long> productStatskedbyDS = productStatsWithWaterMarkDS.keyBy(
                r -> r.getSku_id()
        );

        //TODO 7 开窗
        WindowedStream<ProductStats, Long, TimeWindow> windodDS = productStatskedbyDS.window(TumblingEventTimeWindows.of(Time.seconds(10)));
        //TODO 8 聚合
        SingleOutputStreamOperator<ProductStats> reduceDS = windodDS.reduce(
                new ReduceFunction<ProductStats>() {
                    @Override
                    public ProductStats reduce(ProductStats stats1, ProductStats stats2) throws Exception {
                        stats1.setDisplay_ct(stats1.getDisplay_ct() + stats2.getDisplay_ct());
                        stats1.setClick_ct(stats1.getClick_ct() + stats2.getClick_ct());
                        stats1.setCart_ct(stats1.getCart_ct() + stats2.getCart_ct());
                        stats1.setFavor_ct(stats1.getFavor_ct() + stats2.getFavor_ct());
                        stats1.setOrder_amount(stats1.getOrder_amount().add(stats2.getOrder_amount()));

                        stats1.getOrderIdSet().addAll(stats2.getOrderIdSet());
                        stats1.setOrder_ct(stats1.getOrderIdSet().size() + 0L);

                        stats1.setOrder_sku_num(stats1.getOrder_sku_num() + stats2.getOrder_sku_num());
                        stats1.setPayment_amount(stats1.getPayment_amount().add(stats2.getPayment_amount()));

                        stats1.getRefundOrderIdSet().addAll(stats2.getRefundOrderIdSet());
                        stats1.setRefund_order_ct(stats1.getRefundOrderIdSet().size() + 0L);
                        stats1.setRefund_amount(stats1.getRefund_amount().add(stats2.getRefund_amount()));

                        stats1.getPaidOrderIdSet().addAll(stats2.getPaidOrderIdSet());
                        stats1.setPaid_order_ct(stats1.getPaidOrderIdSet().size() + 0L);

                        stats1.setComment_ct(stats1.getComment_ct() + stats2.getComment_ct());
                        stats1.setGood_comment_ct(stats1.getGood_comment_ct() + stats2.getGood_comment_ct());


                        return stats1;
                    }
                },
                new ProcessWindowFunction<ProductStats, ProductStats, Long, TimeWindow>() {
                    @Override
                    public void process(Long aLong, Context context, Iterable<ProductStats> elements, Collector<ProductStats> out) throws Exception {
                        for (ProductStats productStats : elements) {
                            productStats.setStt(DateTimeUtil.toYMDhms(new Date(context.window().getStart())));
                            productStats.setEdt(DateTimeUtil.toYMDhms(new Date(context.window().getEnd())));
                            out.collect(productStats);
                        }
                    }
                }
        );
        //TODO 9 进行维度关联
        //9.1 通过异步查询，进行维度关联，通过查询维度表中的商品维度信息表
        SingleOutputStreamOperator<ProductStats> withskuDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<ProductStats>("DIM_SKU_INFO") {

                    @Override
                    public String getKey(ProductStats productStats) {
                        return productStats.getSku_id().toString();
                    }

                    @Override
                    public void join(ProductStats productStats, JSONObject dimInfoJsonObj) throws ParseException {
                        productStats.setSku_name(dimInfoJsonObj.getString("SKU_NAME"));
                        productStats.setSku_price(dimInfoJsonObj.getBigDecimal("PRICE"));
                        productStats.setCategory3_id(dimInfoJsonObj.getLong("CATEGORY3_ID"));
                        productStats.setSpu_id(dimInfoJsonObj.getLong("SPU_ID"));
                        productStats.setTm_id(dimInfoJsonObj.getLong("TM_ID"));
                    }
                },
                60,
                TimeUnit.SECONDS

        );
        //9.2 spu
        SingleOutputStreamOperator<ProductStats> withspuDS = AsyncDataStream.unorderedWait(
                withskuDS,
                new DimAsyncFunction<ProductStats>("DIM_SPU_INFO") {

                    @Override
                    public String getKey(ProductStats productStats) {
                        return String.valueOf(productStats.getSpu_id());
                    }

                    @Override
                    public void join(ProductStats productStats, JSONObject dimInfoJsonObj) throws ParseException {
                        productStats.setSpu_name(dimInfoJsonObj.getString("SPU_NAME"));
                    }
                },
                60,
                TimeUnit.SECONDS

        );
        //9.3 品类维度
        SingleOutputStreamOperator<ProductStats> withCateDS = AsyncDataStream.unorderedWait(
                withspuDS,
                new DimAsyncFunction<ProductStats>("DIM_BASE_CATEGORY3") {

                    @Override
                    public String getKey(ProductStats productStats) {
                        return String.valueOf(productStats.getCategory3_id());
                    }

                    @Override
                    public void join(ProductStats productStats, JSONObject dimInfoJsonObj) throws ParseException {
                        productStats.setCategory3_name(dimInfoJsonObj.getString("NAME"));
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        //9.4品牌维度
        SingleOutputStreamOperator<ProductStats> withTmDS = AsyncDataStream.unorderedWait(
                withCateDS,
                new DimAsyncFunction<ProductStats>("DIM_BASE_TRADEMARK") {

                    @Override
                    public String getKey(ProductStats productStats) {
                        return String.valueOf(productStats.getTm_id());
                    }

                    @Override
                    public void join(ProductStats productStats, JSONObject dimInfoJsonObj) throws ParseException {
                        productStats.setTm_name(dimInfoJsonObj.getString("TM_NAME"));

                    }
                },
                60,
                TimeUnit.SECONDS
        );

        withTmDS.print(">>>>>>>");
        //TODO 10 写入到clickhouse中
        withTmDS.addSink(
                ClickHouseUtil.getJdbcSink("insert into product_stats_1021 values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)")
        );
        env.execute();
    }
}
